package com.csw.sql

import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo03DFApi {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .config("spark.sql.shuffle.partitions", 2)
      .master("local")
      .appName("api")
      .getOrCreate()

    //导入隐式转换
    import spark.implicits._

    val studentDF: DataFrame = spark
      .read
      .format("csv")
      .option("seq", ",") //默认是以逗号分隔
      //按照数据的顺序指定表结构
      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
      .load("spark/data/students.txt")

    /**
      * show：查看数据
      */
    studentDF.show()

    /**
      * where：过滤数据
      */
    //字符串表达式
    studentDF.where("age=23").show()

    //使用列表达式的方式
    studentDF.where($"age" === 22).show()


    /**
      * Dataset[Row]  和DataFrame是同一个东西
      */

    /**
      * select：选择数据
      */

    studentDF.select("name", "gender").show()

    //选择的时候也可以处理数据
    studentDF.select($"name", $"age" + 1 as "age").show()

    /**
      * groupBy：分组
      * 分组之后必须接着一个聚合函数
      */

    studentDF.groupBy($"clazz").count().show()


    //导入sql中所有的函数
    import org.apache.spark.sql.functions._

    //计算每个班级最大的年龄
    studentDF
      .groupBy("clazz")
      .agg(max($"age") as "maxAge")
      .show()

    //计算每个班级的平均年龄
    studentDF
      .groupBy("clazz")
      .agg(avg($"age") as "avgAge")
      .show()


    /**
      *
      * join：表关联
      *
      * 如果关联的列名相同直接指定关联列名
      */

    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("seq", ",")
      .schema("sid STRING,subjectId STRING,score DOUBLE")
      .load("spark/data/score.txt")

    val joinDF: DataFrame = studentDF.join(scoreDF, $"id" === $"sid", "inner")

    //如果关联的列名一样直接指定关联列名
    //    studentDF.join(scoreDF, "id")

    joinDF
      .groupBy("name")
      .agg(sum($"score") as "sumScore")
      .show()


  }
}
